package com.yfbao.poll.demo.model;

import cn.hutool.json.JSONUtil;
import com.yfbao.poll.demo.message.BasePollMessage;
import com.yfbao.poll.demo.service.PollingCache;
import com.yfbao.poll.demo.service.PollingExecutor;
import com.yfbao.poll.demo.service.PollingService;
import jakarta.servlet.http.HttpServletResponse;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import javax.servlet.AsyncContext;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;


/**
 * 异步长轮询
 */
public class AsyncLongPoll extends BasePoll implements Runnable{
    private static final Log logger = LogFactory.getLog(AsyncLongPoll.class);

    @Override
    public void run() {
        asyncTimeoutFuture = PollingExecutor.scheduleLongPolling(() -> {
            try {
                logger.info("轮询延时到期，检查是否有触发消息");
                boolean removeFlag = PollingCache.getAllWebMap().remove(userName, AsyncLongPoll.this);
                if(removeFlag){
                    //todo:
                    Map<String, List<BasePollMessage>> userMessagesFromCache = pollService.getUserMessagesFromCache(userMsgTypes, userName);
                    if(userMessagesFromCache.isEmpty()){
                        sendResponse(null);
                    }else {
                        Map<String,Object> resMap =new HashMap<>();
                        resMap.put("Long-poll-timeout",timeoutTime+delayOffset);
                        resMap.put("message",userMessagesFromCache);
                        resMap.put("messageLength",userMessagesFromCache.size());
                        sendResponse(resMap);
                    }

                }else {
                    logger.warn("AsyncLongPolling relations delete fail.");
                }
            } catch (Throwable t) {
                logger.error("AsyncLongPolling error:" + t.getMessage(), t);
                //this.asyncContext.complete();
            }

        }, timeoutTime, TimeUnit.SECONDS);

        //添加只监听队列等待队列
        List<BasePoll> basePolls = PollingCache.getAllWebMap().get(userName);
        if(basePolls ==null){
            basePolls =new CopyOnWriteArrayList<BasePoll>();
            List<BasePoll> olDbasePolls = PollingCache.getAllWebMap().putIfAbsent(userName, basePolls);
            if(olDbasePolls !=null){
                basePolls = olDbasePolls;
            }
        }
        basePolls.add(this);
        PollingCache.getAllWebMap().put(userName,basePolls);
    }

    public void sendResponse(Map<String, Object> changedGroups) {

        // Cancel time out task.
        if (null != asyncTimeoutFuture) {
            asyncTimeoutFuture.cancel(false);
        }
        generateResponse(changedGroups);
    }

    void generateResponse(Map<String,Object> changedGroups) {
        //TODO:
        try {
            if (null == changedGroups) {
                // Tell web container to send http response.
                HttpServletResponse response = (HttpServletResponse)asyncContext.getResponse();
                response.setHeader("Pragma", "no-cache");
                response.setDateHeader("Expires", 0);
                response.setHeader("Cache-Control", "no-cache,no-store");
                response.setStatus(HttpServletResponse.SC_OK);
                response.getWriter().println("no");
                response.getWriter().flush();
                asyncContext.complete();
                return;
            }

            HttpServletResponse response = (HttpServletResponse) asyncContext.getResponse();
            // Disable cache.
            response.setHeader("Pragma", "no-cache");
            response.setDateHeader("Expires", 0);
            response.setHeader("Cache-Control", "no-cache,no-store");
            response.setStatus(HttpServletResponse.SC_OK);
            String res = JSONUtil.toJsonStr(changedGroups);
            response.getWriter().println(res);
            response.getWriter().flush();
            asyncContext.complete();
        } catch (Exception ex) {
            logger.error(ex.toString(), ex);
            asyncContext.complete();
        }
    }

    public AsyncLongPoll(AsyncContext ac, Map<String, Long> userMsgTypes, String userName,
                         long timeoutTime, String tag, PollingService pollService, Integer delayOffset) {
        super(userMsgTypes,userName,timeoutTime,tag,true);
        this.asyncContext = ac;
        this.pollService = pollService;
        this.delayOffset = delayOffset;
    }

    final Integer delayOffset;

    final AsyncContext asyncContext;

    final PollingService pollService;

    //web轮询的消息类型和最后对应的版本号，比如将{msg:[{type:msg1, id:001},{type:msg2,id:010}]}保存至下面map
    //异步schedule执行的句柄
    Future<?> asyncTimeoutFuture;

    @Override
    public String toString() {
        return "AsyncLongPolling{" + "clientListenMsgMap=" + userMsgTypes + ", createTime=" + createTime + ", user='" + userName
                + '\'' +  ", tag='" + tag + '\'' + ", timeoutTime=" + timeoutTime + ",isAsync="+isAsync+'}';
    }
}
